package com.ruoyi.management.mqtt;

import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.management.base.utils.HexUtils;
import com.ruoyi.management.pojo.dto.AmmeterDirectivesDto;
import com.ruoyi.management.pojo.dto.AmmeterResultDto;
import com.ruoyi.management.pojo.param.TidParam;
import com.ruoyi.management.pojo.po.AmmeterAddPO;
import com.ruoyi.management.pojo.po.AmmeterResultPo;
import com.ruoyi.management.pojo.po.AmmeterSwitchOperationPo;
import com.ruoyi.management.pojo.vo.AmmeterResultVO;
import com.ruoyi.management.pojo.vo.AmmeterXQVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

@Slf4j
public class MqttConsumer implements MqttCallbackExtended {

    private final MqttClient mqttClient;
    private final MqttContext mqttContext;
    private final ConcurrentHashMap<String, AmmeterResultPo> ammeterCache = new ConcurrentHashMap<>();

    public MqttConsumer(MqttClient mqttClient, MqttContext mqttContext) {
        this.mqttClient = mqttClient;
        this.mqttContext = mqttContext;
    }

    @Override
    public synchronized void connectComplete(boolean reconnect, String serverURI) {
        log.info("MQTT连接成功: reconnect={}, URI={}", reconnect, serverURI);
        try {
            List<TidParam> tidList = mqttContext.getAmmeterMapper().selectStopic();
            log.info("查询到{}个电表信息", tidList.size());
            if (CollectionUtils.isEmpty(tidList)) {
                log.debug("未查询到任何电表信息");
                return;
            }
            for (TidParam param : tidList) {
                String topic = param.getStopic();
                mqttClient.subscribe(topic);
                log.info("订阅Topic成功: {}", topic);
            }
        } catch (MqttException e) {
            log.error("订阅Topic失败", e);
            throw new RuntimeException("MQTT订阅异常", e);
        }
    }

    @Override
    public synchronized void messageArrived(String topic, MqttMessage message) throws Exception {
        String hexPayload = HexUtils.bytesToHex(message.getPayload());
        if (hexPayload.equals("7777772E7573722E636E") || hexPayload.equals("535A2E444A4B2E303435")) return;
        log.info("收到新电表的消息如下: {}", hexPayload);
//        if (message.isRetained()){
//            String command = new String(message.getPayload());
//            log.info("收到保留消息: {}", command);
//            //handleCommand(command);
//        }else{
//            System.out.println("收到常规消息: " + new String(message.getPayload()));
//        }
        try {
            byte[] data = HexUtils.hexToBytes(hexPayload);
            int deviceAddress = data[0] & 0xFF;
            int functionCode = data[1] & 0xFF;
            int operate = data[4] & 0xFF;
            if (functionCode == 0x06) {
                if (operate == 0xAA) {//拉闸操作
                    AmmeterSwitchOperationPo ammeterSwitchOperationPo = new AmmeterSwitchOperationPo();
                    ammeterSwitchOperationPo.setAmmeterId(String.valueOf(deviceAddress));
                    ammeterSwitchOperationPo.setDbSwitch(0);
                    ammeterSwitchOperationPo.setTopic(topic);
                    mqttContext.getAmmeterMapper().updateOperation(ammeterSwitchOperationPo);
                } else {
                    AmmeterSwitchOperationPo ammeterSwitchOperationPo = new AmmeterSwitchOperationPo();
                    ammeterSwitchOperationPo.setAmmeterId(String.valueOf(deviceAddress));
                    ammeterSwitchOperationPo.setDbSwitch(1);
                    ammeterSwitchOperationPo.setTopic(topic);
                    mqttContext.getAmmeterMapper().updateOperation(ammeterSwitchOperationPo);
                }
                return;
            }

            AmmeterResultDto resultDto = parseDeviceData(HexUtils.hexToBytes(hexPayload), topic);
            String ammeterId = resultDto.getAmmeterId(); // 电表编号

            String directive = mqttContext.getAmmeterMapper().selectAmmeterDirectives
                    (new AmmeterDirectivesDto(ammeterId)
                    );
            log.info("查询指令 -> 电表编号: {}, 电表地址: {} ,操作指令: {}", ammeterId, deviceAddress, directive);

            AmmeterResultPo po = ammeterCache.computeIfAbsent(ammeterId, id -> {
                AmmeterResultPo newPo = new AmmeterResultPo();

                newPo.setDbbh(resultDto.getAmmeterId());
                newPo.setTimestamp(new Date());
                return newPo;
            });

            log.info("po: {}", po);

            // 电表返回的数据单位不同
            String zztt = mqttContext.getAmmeterMapper().selectAmmeterZztt(topic);
            if (zztt.equals("江环项目")) {
                updateAmmeterResultPo(po, directive, resultDto);
            } else {
                // 黄埭项目
                updateAmmeterResultPo1(po, directive, resultDto);
            }

            if (isAllDirectivesProcessed(po)) {
                // mqttContext.getAmmeterMapper().insertData(po);

                /* 每日只存一次电量 */
                // 获取今日日期字符串
//                String today = new SimpleDateFormat("yyyy-MM-dd").format(new Date());
//                String dailySetKey = "ammeter:daily_stored:" + today + ":" + po.getDbbh();
//
//                // 检查该设备今天是否已存储
//                Boolean b = mqttContext.getRedisTemplate().hasKey(dailySetKey);
//                if (Boolean.TRUE.equals(b)) {
//                    ammeterCache.remove(ammeterId);
//                    return;
//                }

                String redisKey = "ammeter:result:" + po.getDbbh() + ":" + new Date();

                mqttContext.getRedisTemplate().opsForValue().set(redisKey, po, 90, TimeUnit.DAYS);

                // 如果电量不为空和0，就记录该设备今天已存储
//                if (ObjectUtils.isNotEmpty(po.getTotalPower()) && !Objects.equals(po.getTotalPower(), BigDecimal.ZERO)) {
//                    mqttContext.getRedisTemplate().opsForValue().set(dailySetKey, po, 48, TimeUnit.HOURS);
//                }

                log.info("数据已缓存至 Redis: ammeterId={}, 过期时间=90天", po.getDbbh());

                ammeterCache.remove(ammeterId);

                log.info("数据聚合完成并保存: ammeterId={}", ammeterId);

            }
        } catch (Exception e) {
            log.error("处理消息异常: topic={}, hex={}", topic, hexPayload, e);
        }
    }

    private void updateAmmeterResultPo(AmmeterResultPo po, String directive, AmmeterResultDto dto) {
        switch (directive) {
            case "1":
                po.setTotalPower(BigDecimal.valueOf(dto.getData() * 0.01));
                break;
            case "3":
                po.setTotalActivePower(dto.getData());
                break;
            case "4":
                po.setVoltageA(dto.getData());
                break;
            case "5":
                po.setVoltageB(dto.getData());
                break;
            case "6":
                po.setVoltageC(dto.getData());
                break;
            case "7":
                po.setCurrentA(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(1000), 3, RoundingMode.HALF_UP));
                break;
            case "8":
                po.setCurrentB(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(1000), 3, RoundingMode.HALF_UP));
                break;
            case "9":
                po.setCurrentC(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(1000), 3, RoundingMode.HALF_UP));
                break;
            case "a":
                po.setPowerFactorA(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(100), 2, RoundingMode.HALF_UP));
                break;
            case "b":
                po.setPowerFactorB(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(100), 2, RoundingMode.HALF_UP));
                break;
            case "c":
                po.setPowerFactorC(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(100), 2, RoundingMode.HALF_UP));
                break;
            case "d":
                po.setFrequencyA(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(100), 2, RoundingMode.HALF_UP));
                break;
            case "e":
                po.setFrequencyB(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(100), 2, RoundingMode.HALF_UP));
                break;
            case "f":
                po.setFrequencyC(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(100), 2, RoundingMode.HALF_UP));
                break;
            case "10":
                po.setTotalReactivePower(dto.getData());
        }
    }

    private void updateAmmeterResultPo1(AmmeterResultPo po, String directive, AmmeterResultDto dto) {
        Long modelId = null; // 电表型号ID
        if (ObjectUtils.isNotEmpty(po)) {
            // 根据电表编号查询电表型号
            AmmeterXQVO ammeterXQVO = mqttContext.getAmmeterMapper().selectAmmeterx(po.getDbbh());
            if (ObjectUtils.isNotEmpty(ammeterXQVO)) {
                modelId = ammeterXQVO.getModelId();
            }
        }
        switch (directive) {
            case "1":
                // 如果电表型号是2或3(海恩德或安科瑞527AMC), 精度是0.001, 其他是0.01
                // 四舍五入保留2位
                if (ObjectUtils.isNotEmpty(modelId)) {
                    if (modelId == 2 || modelId == 3) {
                        po.setTotalPower(BigDecimal.valueOf(dto.getData() * 0.001).setScale(2, RoundingMode.HALF_UP));
                    } else if (modelId == 1 || modelId == 4) {
                        po.setTotalPower(BigDecimal.valueOf(dto.getData() * 0.01).setScale(2, RoundingMode.HALF_UP));
                    } else {
                        po.setTotalPower(BigDecimal.valueOf(dto.getData() * 0.01).setScale(2, RoundingMode.HALF_UP));
                    }
                } else {
                    po.setTotalPower(BigDecimal.valueOf(dto.getData() * 0.01).setScale(2, RoundingMode.HALF_UP));
                }
                break;
            case "3":
                po.setTotalActivePower(dto.getData());
                break;
            case "4":
                po.setVoltageA(dto.getData() / 10);
                break;
            case "5":
                po.setVoltageB(dto.getData() / 10);
                break;
            case "6":
                po.setVoltageC(dto.getData() / 10);
                break;
            case "7":
                po.setCurrentA(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(1000), 3, RoundingMode.HALF_UP));
                break;
            case "8":
                po.setCurrentB(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(1000), 3, RoundingMode.HALF_UP));
                break;
            case "9":
                po.setCurrentC(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(1000), 3, RoundingMode.HALF_UP));
                break;
            case "a":
                po.setPowerFactorA(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(100), 2, RoundingMode.HALF_UP));
                break;
            case "b":
                po.setPowerFactorB(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(100), 2, RoundingMode.HALF_UP));
                break;
            case "c":
                po.setPowerFactorC(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(100), 2, RoundingMode.HALF_UP));
                break;
            case "d":
                po.setFrequencyA(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(100), 2, RoundingMode.HALF_UP));
                break;
            case "e":
                po.setFrequencyB(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(100), 2, RoundingMode.HALF_UP));
                break;
            case "f":
                po.setFrequencyC(BigDecimal.valueOf(dto.getData()).divide(BigDecimal.valueOf(100), 2, RoundingMode.HALF_UP));
                break;
            case "10":
                po.setTotalReactivePower(dto.getData());
        }
    }

    private boolean isAllDirectivesProcessed(AmmeterResultPo po) {
        if (po == null) {
            return false;
        }

        long timeElapsed = System.currentTimeMillis() - po.getTimestamp().getTime();

//        if (timeElapsed > 45) {
//            log.warn("数据采集超时，部分字段可能为null，仍将入库: ammeterId={}", po.getDbbh());
//            return true;
//        }

        return po.getTotalPower() != null
//                && po.getTotalActivePower() != null
//                && po.getVoltageA() != null
//                && po.getVoltageB() != null
//                && po.getVoltageC() != null
//                && po.getCurrentA() != null
//                && po.getCurrentB() != null
//                && po.getCurrentC() != null
//                && po.getPowerFactorA() != null
//                && po.getPowerFactorB() != null
//                && po.getPowerFactorC() != null
//                && po.getFrequencyA() != null
//                && po.getFrequencyB() != null
//                && po.getFrequencyC() != null
//                && po.getTotalReactivePower() != null
                ;
    }


    private BigDecimal scale(double value, double factor) {
        return BigDecimal.valueOf(value * factor);
    }

    //========================= 缓存清理 =========================//
    @Scheduled(fixedRate = 60000)
    public void cleanupExpiredCache() {
        long now = System.currentTimeMillis();
        ammeterCache.entrySet().removeIf(entry ->
                (now - entry.getValue().getTimestamp().getTime()) > 60000
        );
    }

    private AmmeterResultDto parseDeviceData(byte[] data, String topic) {

        if (data.length < 7) {
            log.debug("数据长度不足");
        }


        int deviceAddress = data[0] & 0xFF;
        String dbbh = mqttContext.getAmmeterMapper().selectDbbh(deviceAddress, topic); // 电表编号
        // 查询电表信息
        AmmeterAddPO ammeterAddPO = mqttContext.getAmmeterMapper().selectdbbh(dbbh);
        Long modelId = null;
        if (ObjectUtils.isNotEmpty(ammeterAddPO)) {
            modelId = ammeterAddPO.getModelId();
        }

        int functionCode = data[1] & 0xFF;
        int dataLength = data[2] & 0xFF;
        int operate = data[4] & 0xFF;

        int expectedLength = 3 + dataLength + 2;
        if (data.length != expectedLength) {
            log.debug("数据长度不符，预期长度: {}，实际长度: {}", expectedLength, data.length);
        }

        byte[] dataContent = Arrays.copyOfRange(data, 3, 3 + dataLength);
        byte[] receivedCRC = Arrays.copyOfRange(data, data.length - 2, data.length);

        byte[] calculatedCRC = HexUtils.calculateCRC(Arrays.copyOfRange(data, 0, data.length - 2));
        log.info("topic: {}, 数据内容: {}, CRC: {}, 计算的CRC: {}", topic, HexUtils.bytesToHex(dataContent), HexUtils.bytesToHex(receivedCRC), HexUtils.bytesToHex(calculatedCRC));

        if (!Arrays.equals(receivedCRC, calculatedCRC)) {
            log.debug("CRC校验失败");
        }

        // 如果电表型号是2-安科瑞527AMC
        if (ObjectUtils.isNotEmpty(modelId) && 2 == modelId) {
            // 安科瑞527AMC电表使用单精度浮点数格式
            if (dataLength == 4) {
                // 将4字节数据转换为单精度浮点数
                int intValue = ((dataContent[0] & 0xFF) << 24) |
                        ((dataContent[1] & 0xFF) << 16) |
                        ((dataContent[2] & 0xFF) << 8) |
                        (dataContent[3] & 0xFF);

                float floatValue = Float.intBitsToFloat(intValue);

                AmmeterResultDto ammeterResultDto = new AmmeterResultDto();
                ammeterResultDto.setAmmeterId(dbbh);
                ammeterResultDto.setData((int) floatValue);
                return ammeterResultDto;

            } else if (dataLength == 2) {
                // 2字节数据，保持原有逻辑
                int registerValue = ((dataContent[0] & 0xFF) << 8) | (dataContent[1] & 0xFF);

                AmmeterResultDto ammeterResultDto = new AmmeterResultDto();
                ammeterResultDto.setAmmeterId(dbbh);
                ammeterResultDto.setData(registerValue);
                return ammeterResultDto;
            } else if (dataLength == 6) {
                // 6字节数据，保持原有逻辑
                // 注意：这里需要处理long类型，因为6字节可能超过int范围
                long registerValue = ((dataContent[0] & 0xFFL) << 40) |
                        ((dataContent[1] & 0xFFL) << 32) |
                        ((dataContent[2] & 0xFFL) << 24) |
                        ((dataContent[3] & 0xFFL) << 16) |
                        ((dataContent[4] & 0xFFL) << 8) |
                        (dataContent[5] & 0xFFL);

                AmmeterResultDto ammeterResultDto = new AmmeterResultDto();
                ammeterResultDto.setAmmeterId(dbbh);
                ammeterResultDto.setData((int) registerValue);
                return ammeterResultDto;
            } else {
                // 默认处理2字节数据
                int registerValue = ((dataContent[0] & 0xFF) << 8) | (dataContent[1] & 0xFF);

                AmmeterResultDto ammeterResultDto = new AmmeterResultDto();
                ammeterResultDto.setAmmeterId(dbbh);
                ammeterResultDto.setData(registerValue);
                return ammeterResultDto;
            }
        } else {
            if (dataLength == 4) {
                int registerValue = ((dataContent[0] & 0xFF) << 24) |
                        ((dataContent[1] & 0xFF) << 16) |
                        ((dataContent[2] & 0xFF) << 8) |
                        (dataContent[3] & 0xFF);

                AmmeterResultDto ammeterResultDto = new AmmeterResultDto();
                ammeterResultDto.setAmmeterId(dbbh);
                ammeterResultDto.setData(registerValue);
                return ammeterResultDto;

            } else if (dataLength == 2) {
                int registerValue = ((dataContent[0] & 0xFF) << 8) | (dataContent[1] & 0xFF);

                AmmeterResultDto ammeterResultDto = new AmmeterResultDto();
                ammeterResultDto.setAmmeterId(dbbh);
                ammeterResultDto.setData(registerValue);
                return ammeterResultDto;
            } else if (dataLength == 6) {
                int registerValue = ((dataContent[0] & 0xFF) << 40) |
                        ((dataContent[1] & 0xFF) << 32) |
                        ((dataContent[2] & 0xFF) << 24) |
                        ((dataContent[3] & 0xFF) << 16) |
                        ((dataContent[4] & 0xFF) << 8) |
                        ((dataContent[5] & 0xFF));

                AmmeterResultDto ammeterResultDto = new AmmeterResultDto();
                ammeterResultDto.setAmmeterId(dbbh);
                ammeterResultDto.setData(registerValue);
                return ammeterResultDto;
            } else {
                int registerValue = ((dataContent[0] & 0xFF) << 8) | (dataContent[1] & 0xFF);

                AmmeterResultDto ammeterResultDto = new AmmeterResultDto();
                ammeterResultDto.setAmmeterId(dbbh);
                ammeterResultDto.setData(registerValue);
                return ammeterResultDto;
            }
        }
    }

    @Override
    public void connectionLost(Throwable cause) {
        log.error("MQTT连接丢失,原因:{}", cause.getMessage(), cause);

        try {
            if (mqttClient != null && !mqttClient.isConnected()) {
                log.info("尝试重新连接MQTT服务器...");
                mqttClient.connect();
                log.info("重新连接成功");
            }
        } catch (MqttException e) {
            log.error("重新连接失败,原因:{}", e.getMessage(), e);

        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("消息发送完成, token: {}", token);
    }

    //========================= 生命周期管理 =========================//
    @PreDestroy
    public void destroy() {
        try {
            if (mqttClient != null && mqttClient.isConnected()) {
                mqttClient.disconnect();
                log.info("MQTT客户端已断开连接");
            }
        } catch (MqttException e) {
            log.error("断开MQTT客户端连接时发生异常", e);
        }
    }


}
